package com.tous.modules.handler;

import com.alibaba.fastjson2.JSON;
import com.tous.modules.bean.AckMessage;
import com.tous.modules.bean.UdpMessage;
import com.tous.modules.service.IUdpBusinessService;
import io.micrometer.core.annotation.Counted;
import io.micrometer.core.annotation.Timed;
import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Timer;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.socket.DatagramPacket;
import jakarta.annotation.Resource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author mengwei
 * @description UdpServerHandler
 * @createDate 2025/7/16 15:20
 */
@Component
@ChannelHandler.Sharable
public class UdpServerHandler extends SimpleChannelInboundHandler<DatagramPacket> {


    public void setApplicationContext(ApplicationContext applicationContext) {
        this.applicationContext = applicationContext;
    }

    private ApplicationContext applicationContext;

    private static final Logger logger = LoggerFactory.getLogger(UdpServerHandler.class);
    private final AtomicInteger activeConnections = new AtomicInteger();
    private final MeterRegistry meterRegistry;
    private final Counter errorCounter;
    private final Timer processingTimer;

    // 已接收消息ID记录，用于去重
    private final ConcurrentHashMap<Long, Boolean> receivedMessages = new ConcurrentHashMap<>();

    public UdpServerHandler(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;

        // 注册连接指标
        meterRegistry.gauge("udp.server.active_connections", activeConnections);

        // 注册错误计数器
        errorCounter = Counter.builder("udp.server.errors")
                .description("Number of errors in UDP server")
                .tag("type", "general")
                .register(meterRegistry);

        // 注册处理时间计时器
        processingTimer = Timer.builder("udp.server.processing_time")
                .description("Time taken to process UDP messages")
                .publishPercentiles(0.5, 0.95, 0.99)
                .register(meterRegistry);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        activeConnections.incrementAndGet();
        ctx.fireChannelActive();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        activeConnections.decrementAndGet();
        ctx.fireChannelInactive();
    }

    @Override
    @Timed(value = "udp.request.processing.time", description = "UDP请求处理时间")
    @Counted(value = "udp.request.count", description = "UDP请求计数")
    protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket packet) throws Exception {
        processingTimer.record(() -> {
            try {
                String msg = packet.content().toString(StandardCharsets.UTF_8);
                logger.info("接收到UDP消息: {}", msg);
                // 记录消息大小
                UdpMessage udpMessage = JSON.parseObject(msg, UdpMessage.class);
                meterRegistry.summary("udp.message.size")
                        .record(msg.length());

                if (udpMessage.getNeedAck() == 1) {
                    handleUdpMessage(ctx, packet, udpMessage);
                }else {
                    handleAckMessage(udpMessage);
                }


            } catch (Exception e) {
                errorCounter.increment();
                logger.error("处理UDP消息时出错", e);
                ctx.close();
            }
        });
    }




    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        errorCounter.increment();
        logger.error("UDP处理异常", cause);
        ctx.close();
    }

    private void handleUdpMessage(ChannelHandlerContext ctx, DatagramPacket packet, UdpMessage message) {
        Long messageId = message.getMessageId();
        InetSocketAddress sender = packet.sender();

        // 检查消息是否已处理过
        if (receivedMessages.putIfAbsent(messageId, true) != null) {
            logger.info("重复消息，已忽略: {}", messageId);
            // 重新发送确认，以防客户端未收到
            sendAck(ctx, messageId, sender, true, null);
            return;
        }

        logger.info("收到新消息 [{}]: {}", messageId, message.getContent());
        UdpMessage response = UdpMessage.error();
        try {
            // 处理消息 (这里可以添加业务逻辑)
            response = processMessage(message);
            // 如果需要确认，发送ACK
            if (message.getNeedAck() ==1) {
                sendAck(ctx, messageId, sender, true, response);
            }
        } catch (Exception e) {
            logger.error("处理消息失败: {}", messageId, e);
            if (message.getNeedAck() ==1) {
                sendAck(ctx, messageId, sender, false, null);
            }
        }
    }

    private void handleAckMessage(UdpMessage ack) {
        logger.info("收到确认消息: {}", ack);
    }

    private void sendAck(ChannelHandlerContext ctx, Long messageId, InetSocketAddress recipient
            , boolean success, UdpMessage message) {
        AckMessage ack = new AckMessage();
        ack.setMessageId(messageId);
        ack.setSuccess(success ? 1 : 0);
        ack.setUdpMessage(message);
        ctx.writeAndFlush(new DatagramPacket(
                Unpooled.copiedBuffer(JSON.toJSONString(ack), StandardCharsets.UTF_8),
                recipient));
        logger.info("发送确认消息: {}", ack);
    }

    private UdpMessage processMessage(UdpMessage message) {
        // 模拟业务处理
        UdpMessage response = UdpMessage.error();
        try {
            IUdpBusinessService udpBusinessService = applicationContext.getBean(IUdpBusinessService.class);
            response  = udpBusinessService.handler(message);
        } catch (Exception e) {
            Thread.currentThread().interrupt();
            logger.error("消息处理失败: {}", message.getMessageId(), e);
            response.setContent(e.getMessage());
        }
        logger.info("消息处理完成: {}", message.getMessageId());
        return response;
    }

}
